package io.yunho.srw.lean.stream.concurrent;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @author: Jason
 * @date: 2021/5/8 20:10
 * @description:
 */
public class StreamConcurrentDemo {
    public static void main(String[] args) {
        //当前是单行输出的
        IntStream range = IntStream.range(1, 100);
        //long count1 = range.peek(StreamConcurrentDemo::debug).count();
      //  System.out.println(count1);
        //parallel方法会产生并行流，从而使各线程同时处理该流中的各元素
//        long count = range.parallel().peek(StreamConcurrentDemo::debug).count();
//        System.out.println(count);
        //实现一个先并行再串行的流
        //实际并未实现该效果，结论是先后调用的话，以最后的方法为准，当前示例中为串行
//        long count = range
//                //
//                .parallel().peek(StreamConcurrentDemo::debug)
//                //sequential方法会产生串行流
//                .sequential().peek(StreamConcurrentDemo::debug2)
//                .count();
//        System.out.println(count);

        //查看并行流的线程情况 默认使用的线程池：ForkJoinPool.commonPool（JDK自带） 默认线程数当前机器的CPU个数
        //可以手动修改线程数
//        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","20");
//        long count = range.parallel().peek(StreamConcurrentDemo::threaddebug).count();

        //自定义并行流线程池  ForkJoinPool-1
        ForkJoinPool forkJoinPool = new ForkJoinPool(20);
        forkJoinPool.submit(()->range.parallel().peek(StreamConcurrentDemo::threaddebug).count());
        forkJoinPool.shutdown();

        synchronized (forkJoinPool){//傻等线程完成 避免主线程直接执行完成退出
            try {
                forkJoinPool.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void debug(int i){
        System.out.println("debug: "+i);
        try {
            TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}
    }
    public static void debug2(int i){
        System.err.println("debug2: "+i);
        try {
            TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}
    }

    public static void threaddebug(int i){
        System.out.println(Thread.currentThread().getName()+"******threaddebug: "+i);
        try {
            TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}
    }
}
